package com.zqh.kafka.demo.producer;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class MyProducer1 {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        Map<String, Object> configs = new HashMap<>();
        // 添加配置参数
        configs.put("bootstrap.servers", "centos122:9092"); // 服务地址
        configs.put("key.serializer", IntegerSerializer.class); // 指定key的序列化类
        configs.put("value.serializer", StringSerializer.class); // 指定value的序列化类

        configs.put("acks", "all");
        configs.put("reties", 3); // 重试三次

        // 指定 key 和 value 的类型
        KafkaProducer<Integer, String> producer = new KafkaProducer<Integer, String>(configs);

        // 自定义消息头
        List<Header> headers = new ArrayList<>();
        headers.add(new RecordHeader("biz.name", "producer.demo".getBytes()));

        // 定义消息内容
        ProducerRecord<Integer, String> record = new ProducerRecord<Integer, String>(
                "topic_1",
                0,
                0,
                "hello 0",
                headers
        );
/*
        // 发送   消息的同步确认方式
        Future<RecordMetadata> future = producer.send(record);
        // 获取结果
        RecordMetadata metadata = future.get();

        System.out.println(metadata.topic());
        System.out.println(metadata.partition());
        System.out.println(metadata.offset());

 */
        // 消息的异步确认
        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception e) {
                if (e == null) {
                    System.out.println(metadata.topic());
                    System.out.println(metadata.partition());
                    System.out.println(metadata.offset());
                } else {
                    System.out.println("异常消息：" + e.getMessage());
                }
            }
        });

        // 关闭生产者
        producer.close();

    }

}
